package cn.tworice.netty.message;


import cn.tworice.common.util.StringUtils;
import cn.tworice.netty.constand.MessageConst;
import cn.tworice.netty.exception.NotExistException;
import cn.tworice.netty.session.Session;
import cn.tworice.netty.session.SessionManager;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONException;
import com.alibaba.fastjson.JSONObject;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Optional;
import java.util.function.Consumer;
import java.util.function.Function;

@Component
@Slf4j
public class MessageManager {

    @Resource
    private SessionManager sessionManager;
    /**
     * 收到消息后执行的方法对象
     * 生产环境使用需要将此方法替换为自己的处理方法
     **/
    private Consumer<? super Message> messageConsumer = this::sendMessage;

    /**
     * 客户端下线后执行的方法对象
     */
    private Consumer<? super Session> offLineConsumer = this::offLineMessage;

    /**
     * 客户端下线后执行的方法对象
     */
    private Consumer<? super Session> onLineConsumer = this::onLineMessage;

    /**
     * 认证方法对象
     * 入参：收到的Message
     * 出参：认证后的用户标识
     * 生产环境使用需要将此方法替换为严格认证流程方法
     **/
    private Function<? super Message, String> authFunction = Message::getUid;

    /**
     * 设置收到消息后的处理方法
     * @param consumer 处理方法
     **/
    public MessageManager setReceiveConsumer(Consumer<Message> consumer) {
        this.messageConsumer = consumer;
        return this;
    }

    /**
     * 设置客户端下线消息处理方法
     * @param consumer 处理方法
     * @return 当前消息处理器
     */
    public MessageManager setOffLineConsumer(Consumer<? super Session> consumer) {
        this.offLineConsumer = consumer;
        return this;
    }

    /**
     * 设置客户端上线消息处理方法
     * @param consumer 处理方法
     * @return 当前消息处理器
     */
    public MessageManager setOnLineConsumer(Consumer<? super Session> consumer) {
        this.onLineConsumer = consumer;
        return this;
    }

    /**
     * 设置用户验证处理方法
     * @param authFunction 处理方法
     **/
    public MessageManager setAuthFunction(Function<? super Message, String> authFunction) {
        this.authFunction = authFunction;
        return this;
    }

    /**
     * 发送消息给某个人
     *
     * @param uid 用户标识
     * @param msg 消息体
     **/
    public void sendUid(String uid, String msg) {
        if (StringUtils.isEmpty(uid)) {
            throw new NotExistException("TO_ONE模式下用户标识不能为空");
        }
        log.debug("发送消息给用户:{}",uid);
        if (sessionManager.getUidGroup(uid).isEmpty()) {
            throw new NotExistException(uid + "不存在");
        }
        sessionManager.getUidGroup(uid)
                .writeAndFlush(new TextWebSocketFrame(msg));
    }

    /**
     * 如果目标通道存在则推送消息
     * @param uid 用户编号
     * @param msg 消息内容
     */
    public void sendUidIfExist(String uid, String msg) {
        if(sessionManager.getUidGroup(uid)!=null){
            this.sendUid(uid, msg);
        }
    }

    /**
     * 发送消息给所有在线的人
     *
     * @param msg 消息体
     **/
    public void sendAll(String msg) {
        log.debug("发送消息给所有人");
        sessionManager.getAllChannelGroup()
                .writeAndFlush(new TextWebSocketFrame(msg));
    }

    /**
     * 向组内所有客户端发送消息
     *
     * @param gid 组标识
     * @param msg 消息体
     **/
    public void sendGid(String gid, String msg) {
        if (StringUtils.isEmpty(gid)) {
            throw new NotExistException("TO_GROUP模式下组标识不能为空");
        }
        log.debug("发送消息给组:{}",gid);
        Optional.ofNullable(sessionManager.getGidGroup(gid))
                .ifPresent(channels ->
                        channels.writeAndFlush(new TextWebSocketFrame(msg))
                );
    }

    /**
     * 向组内所有客户端发送消息
     *
     * @param msg 消息体
     **/
    public void sendGid(Message msg) {
        log.debug("发送消息给组:{}",msg.getTo());
        JSONObject message = new JSONObject();
        message.put("form", msg.getUid());
        message.put("message", msg.getContent());
        message.put("format", msg.getFormat());
        Optional.ofNullable(sessionManager.getGidGroup(msg.getTo()))
                .ifPresent(channels ->
                        channels.writeAndFlush(new TextWebSocketFrame(message.toJSONString()))
                );
    }

    /**
     * 发送消息
     * @param message 消息体
     **/
    public void sendMessage(Message message) {
        if (Message.MessageType.TO_ONE.equals(message.getType())) {
            // 某个用户
            this.sendUid(message.getTo(), message.getContent());
        } else if (Message.MessageType.TO_ALL.equals(message.getType())) {
            // 所有客户端
            this.sendAll(message.getContent());
        } else if (Message.MessageType.TO_GROUP.equals(message.getType())) {
            // 某个组
            this.sendGid(message);
        }
    }

    /**
     * 将收到消息的操作公开出去
     * @param ctx   客户端
     * @param frame 消息体
     **/
    public void receiveMessage(ChannelHandlerContext ctx, TextWebSocketFrame frame) {
        try {
            Optional.ofNullable(JSON.parseObject(frame.text(), Message.class)).ifPresent(message -> {
                if (Message.MessageType.LOGIN.equals(message.getType())) {
                    // 认证方法
                    String uid = this.authFunction.apply(message);
                    if (!StringUtils.isEmpty(uid)) {
                        // 认证成功 将信息添加到客户端中
                        Session.set(MessageConst.uid, uid, ctx.channel());
                        Optional.ofNullable(message.getGid()).ifPresent(gid ->
                            Session.set(MessageConst.gid, gid, ctx.channel())
                        );
                        this.onLine(sessionManager.addChannel(ctx.channel()));
                    }
                } else if (!StringUtils.isEmpty(Session.get(MessageConst.uid, ctx.channel()))) {
                    // 手动设置 将认证人覆盖到发送人
                    message.setUid(Session.get(MessageConst.uid, ctx.channel()));
                    this.messageConsumer.accept(message);
                } else {
                    // 未经过认证也不认证就关掉
                    ctx.channel().close();
                }
            });
        } catch (JSONException exception) {
            throw new RuntimeException("收到的消息无法格式化");
        }
    }

    /**
     * 客户端下线
     * @param session 客户端Session
     */
    public void offLine(Session session) {
        this.offLineConsumer.accept(session);
    }

    /**
     * 客户端上线
     * @param session 客户端Session
     */
    public void onLine(Session session) {
        this.onLineConsumer.accept(session);
    }

    public void offLineMessage(Session session) {
        log.debug("客户端下线：{}", session.getChannel().id().asLongText());
    }

    public void onLineMessage(Session session) {
        log.debug("客户端上线：{}", session.getChannel().id().asLongText());
    }
}
